import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;


public class ii {
    public static Configuration configuration; //HBase 配置信息
    public static Connection connection; //HBase 连接
    public static Admin admin;
    public static void main (String [] args) throws IOException{
        init();//初始化
        //建立表
        //deleteTable("student");
        //deleteTable("course");
        //deleteTable("Scourse");


        System.out.println("start to create table!");
        createTable("student", new String[]{"S_Information"});
        createTable("course", new String[]{"C_Information"});
        createTable("Scourse", new String[]{"C_Score"});
        //插入学生表的数据
        insertData("student","2015001","S_Information","name","Li Lei");
        insertData("student","2015001","S_Information","sex","male");
        insertData("student","2015001","S_Information","age","23");
        //insertData("student","2015001","S_Score","123001","86");
        //insertData("student","2015001","S_Score","123003","69");

        insertData("student","2015002","S_Information","name","Han Meimei");
        insertData("student","2015002","S_Information","sex","female");
        insertData("student","2015002","S_Information","age","22");
        //insertData("student","2015002","S_Score","123002","77");
        //insertData("student","2015002","S_Score","123003","99");

        insertData("student","2015003","S_Information","name","Zhangsan");
        insertData("student","2015003","S_Information","sex","male");
        insertData("student","2015003","S_Information","age","24");
        //insertData("student","2015003","S_Score","123002","95");
        //insertData("student","2015003","S_Score","123001","98");


        //插入课程表的数据
        insertData("course","123001","C_Information","name","Math");
        insertData("course","123001","C_Information","credit","2.0");
        insertData("course","123002","C_Information","name","Computer Science");
        insertData("course","123002","C_Information","credit","5.0");
        insertData("course","123003","C_Information","name","English");
        insertData("course","123003","C_Information","credit","3.0");


        //插入选课表的数据
        insertData("Scourse","1230012015001","C_Score","score","86");
        insertData("Scourse","1230012015003","C_Score","score","69");
        insertData("Scourse","1230022015002","C_Score","score","77");
        insertData("Scourse","1230022015003","C_Score","score","99");
        insertData("Scourse","1230032015001","C_Score","score","98");
        insertData("Scourse","1230032015002","C_Score","score","95");

        queryTable("student");
        queryTable("course");
        queryTable("Scourse");
        //查询选修computer science同学的数据
        //List<Map.Entry> listscore=getRowRangeData("Scourse","1230022015001", "1230022015003");
        getData("Scourse","1230022015002","C_Score","score");
        getData("Scourse","1230022015003","C_Score","score");
        //add new columnfamily
        addColumnFamily("student", "Contact");
        //插入contact:email的数据
        insertData("student","123001","Contact","email","lilei@qq.com");
        insertData("student","123002","Contact","email","hmm@@qq.com");
        insertData("student","123003","Contact","email","zs@qq.com");
        //query table
        queryTable("student");
        //queryTable("course");
        //queryTable("Scourse");
        //删除选课记录
        deleteRow("Scourse","1230012015003" );
        deleteRow("Scourse","1230022015003" );
        //scourse
        queryTable("Scourse");

        //清除数据
        deleteTable("student");
        deleteTable("course");
        deleteTable("Scourse");
        close();//关闭连接
    }
    //建立连接
    public static void init () {
        //System.out.println("1!");
        configuration = HBaseConfiguration.create();
        //System.out.println("2!");
        //configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        //System.out.println("3!");
        configuration.set("hbase.zookeeper.quorum","localhost");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            //System.out.println("4!");
            admin = connection.getAdmin();
            //System.out.println("5!");
        }catch(IOException e){
            e.printStackTrace();
            //System.out.println("6!");
        }
    }

    //关闭连接
    public static void close() {
        try{
            if(admin != null) {
                admin.close();
            }
            if ( connection!= null) {
                connection.close();
            }
        }catch (IOException e) {
            e.printStackTrace();
        }
    }
    //创建表
    public static void createTable(String myTableName,String[] colFamily) throws IOException {
        //System.out.println("1");
        TableName tableName = TableName.valueOf(myTableName);
        //System.out.println("2");
        //TableName table = TableName.valueOf(myTableName);
        /**/
        if(admin.tableExists(tableName)){
            System.out.println("Table exists!");
        }
        else {
            //System.out.println("3");
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
            System.out.println("add colfamily");
            for(String str:colFamily){
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(str);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("Create table successfully!");
        }

    }
    //添加课程成绩
    public static void insertData(String tableName, String rowKey, String colFamily, String col, String val) throws IOException {
        System.out.println("insert data");
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(rowKey.getBytes());
        put.addColumn(colFamily.getBytes(),col.getBytes(),val.getBytes());
        table.put(put);
        table.close();
        System.out.println(tableName+" "+rowKey+" "+colFamily+" "+col+val+" add successfully");
    }
    //新建columnfamily
    public static void addColumnFamily(String tableNameString, String columnFamily) throws IOException {
        System.out.println("新建列簇开始");
        //取得目标数据表的标明对象
        TableName tableName = TableName.valueOf(tableNameString);
        //创建列簇对象
        HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
        //将新建的加入到指定的数据表
        admin.addColumn(tableName, columnDescriptor);
        System.out.println("新建列簇结束");
    }
    //浏览成绩
    public static void getData (String tableName,String rowKey,String colFamily,String col)throws IOException {
        System.out.println("getdata");
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get (rowKey. getBytes());
        get.addColumn(colFamily.getBytes(),col.getBytes());
        Result result = table.get(get);
        System.out.println(new String(result.getValue(colFamily.getBytes(),
                col.getBytes())));
        table.close();
    }
    //查詢整個表的數據
    public static void queryTable(String tableNameString) throws IOException {
        System.out.println("--------------------查询整表的数据--------");
        //获取数据表对象
        Table table = connection.getTable(TableName.valueOf(tableNameString));
        //获取表中的数据
        ResultScanner scanner = table.getScanner(new Scan());
        //循环输出表中的数据
        for (Result result : scanner) {
            byte[] row = result.getRow();
            System.out.println("row key is:"+new String(row));
            List<Cell> listCells = result.listCells();
            for (Cell cell : listCells) {
                byte[] familyArray = cell.getFamilyArray();
                byte[] qualifierArray = cell.getQualifierArray();
                byte[] valueArray = cell.getValueArray();
                System.out.println("row value is:"+ new String(familyArray) +
                        new String(qualifierArray) + new String(valueArray));
            }
        }
        System.out.println("---------------查询整表数据结束----------");
    }
    //按照范围查询数据
    public static List<Map.Entry> getRowRangeData(String tableName,String startRowKey, String endRowKey) throws IOException{
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowKey));
        scan.setStopRow(Bytes.toBytes(endRowKey));
        ResultScanner results =  table.getScanner(scan);
        Map<String,Result> map = new HashMap<String,Result>();
        for (Result result : results){
            String key = new String(result.getRow());
            map.put(key,result);
            System.out.println(key +" : "+result);
        }
        List<Map.Entry> list = new ArrayList<Map.Entry>();
        list.addAll(map.entrySet());
        table.close();
        System.out.println("return successfully");
        return list;

    }
    //删除某一指定的行
    public static void deleteRow(String tableName, String rowKey) throws IOException {
        System.out.println("-------------------------清空某一行------------------");
        //取得目标数据表的表明对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete=new Delete(rowKey.getBytes());
        table.delete(delete);
        table.close();
        System.out.println("-------------------------清空结束-----------------");
    }
    //使表失效
    public static void truncateTable(String tableNameString) throws IOException {
        System.out.println("-------------------------清空表开始------------------");
        //取得目标数据表的表明对象
        TableName table = TableName.valueOf(tableNameString);
        //Table table = connection.getTable(TableName.valueOf(tableNameString));
        //设置表状态为无效
        admin = connection.getAdmin();
        admin.disableTable(table);
        //清空指定表的数据
        admin.truncateTable(table, true);
        System.out.println("-------------------------清空表结束-----------------");
    }
    //清空表
    public static void deleteTable(String tableNameString) throws IOException {
        System.out.println("-------------------------刪除表开始------------------");
        //取得目标数据表的表明对象
        TableName table = TableName.valueOf(tableNameString);
        //Table table = connection.getTable(TableName.valueOf(tableNameString));
        //设置表状态为无效
        admin = connection.getAdmin();
        admin.disableTable(table);
        //清空指定表的数据
        admin.deleteTable(table);
        System.out.println("-------------------------刪除表结束-----------------");
    }

}